package com.kedacom.demo.springrabbitex;

import java.util.UUID;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.stereotype.Component;

/**
 * Rabbit Message Sender<br />
 * 封装了需要同步等待ack返回的发送RMQ消息接口，如果等待ack超时或返回nack则抛出NackException<br />
 * @author luocanfeng
 * @date 2013-8-20
 */
@Component("rabbitMessageSender")
public class RabbitMessageSender {

	protected Logger logger = LoggerFactory.getLogger(getClass());

	@Resource
	private RabbitTemplate rabbitTemplate;
	private static final int DEFAULT_CONFIRM_TIMEOUT = 5000;
	private int confirmTimeout = DEFAULT_CONFIRM_TIMEOUT;

	@Resource
	private PendingConfirmHolder confirmHolder;

	public void setRabbitTemplate(RabbitTemplate rabbitTemplate) {
		this.rabbitTemplate = rabbitTemplate;
	}

	public void setConfirmHolder(PendingConfirmHolder confirmHolder) {
		this.confirmHolder = confirmHolder;
	}

	public int getConfirmTimeout() {
		return confirmTimeout;
	}

	public void setConfirmTimeout(int confirmTimeout) {
		this.confirmTimeout = confirmTimeout;
	}

	/**
	 * 发送RMQ消息。
	 */
	public void convertAndSend(Object message) throws AmqpException {
		this.convertAndSend(null, null, message, false);
	}

	/**
	 * 发送RMQ消息。
	 */
	public void convertAndSend(String routingKey, Object message) throws AmqpException {
		this.convertAndSend(null, routingKey, message, false);
	}

	/**
	 * 发送RMQ消息。
	 */
	public void convertAndSend(String exchange, String routingKey, Object message) throws AmqpException {
		this.convertAndSend(exchange, routingKey, message, false);
	}

	/**
	 * 发送RMQ消息。
	 * @param waitForAck 是否同步等待返回ack。
	 */
	public void convertAndSend(Object message, boolean waitForAck) throws AmqpException {
		this.convertAndSend(null, null, message, waitForAck);
	}

	/**
	 * 发送RMQ消息。
	 * @param waitForAck 是否同步等待返回ack。
	 */
	public void convertAndSend(String routingKey, Object message, boolean waitForAck) throws AmqpException {
		this.convertAndSend(null, routingKey, message, waitForAck);
	}

	/**
	 * 发送RMQ消息。
	 * @param waitForAck 是否同步等待返回ack。
	 */
	private void convertAndSend(String exchange, String routingKey, Object message, boolean waitForAck)
			throws AmqpException {
		if (waitForAck) {
			this.convertSendAndWaitForAck(exchange, routingKey, message);
		} else {
			if (StringUtils.isNotEmpty(exchange) && StringUtils.isNotEmpty(routingKey)) {
				rabbitTemplate.convertAndSend(exchange, routingKey, message);
			} else if (StringUtils.isNotEmpty(routingKey)) {
				rabbitTemplate.convertAndSend(routingKey, message);
			} else {
				rabbitTemplate.convertAndSend(message);
			}
		}
	}

	/**
	 * 发送RMQ消息，并同步等待返回ack。
	 */
	private void convertSendAndWaitForAck(String exchange, String routingKey, Object message) throws AmqpException {
		String msgTag = UUID.randomUUID().toString();
		CorrelationData correlationData = new CorrelationData(msgTag);
		PendingConfirm pendingConfirm = new PendingConfirm(msgTag);
		confirmHolder.put(msgTag, pendingConfirm);

		if (StringUtils.isNotEmpty(exchange) && StringUtils.isNotEmpty(routingKey)) {
			rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
		} else if (StringUtils.isNotEmpty(routingKey)) {
			rabbitTemplate.convertAndSend(routingKey, message, correlationData);
		} else {
			rabbitTemplate.correlationconvertAndSend(message, correlationData);
		}

		this.waitForAck(pendingConfirm);
	}

	private void waitForAck(PendingConfirm pendingConfirm) throws NackException {
		LinkedBlockingQueue<Boolean> replyHandoff = pendingConfirm.getQueue();
		try {
			// receive the ack
			Boolean ack = (confirmTimeout < 0) ? replyHandoff.take() : replyHandoff.poll(confirmTimeout,
					TimeUnit.MILLISECONDS);
			if (ack == null) {
				// timeout
				throw new NackException("Timeout while waiting for ack.");
			}

			logger.debug("receive {} for message with correlation '{}'", ack ? "ack" : "nack",
					pendingConfirm.getSavedCorrelation());

			// remove the pending confirm from confirm holder
			confirmHolder.remove(pendingConfirm.getSavedCorrelation());

			if (!ack) {
				// process nack, throw exception
				throw new NackException("Receive a nack.");
			}
		} catch (InterruptedException e) {
			throw new NackException(e);
		}
	}

	/**
	 * 发送RPC RMQ消息，并返回PPC调用结果。
	 */
	public <T> T convertSendAndReceive(Object message) throws AmqpException {
		return this.convertSendAndReceive(null, null, message);
	}

	/**
	 * 发送RPC RMQ消息，并返回PPC调用结果。
	 */
	public <T> T convertSendAndReceive(String routingKey, Object message) throws AmqpException {
		return this.convertSendAndReceive(null, null, message);
	}

	/**
	 * 发送RPC RMQ消息，并返回PPC调用结果。
	 */
	@SuppressWarnings("unchecked")
	private <T> T convertSendAndReceive(String exchange, String routingKey, Object message) throws AmqpException {
		if (StringUtils.isNotEmpty(exchange) && StringUtils.isNotEmpty(routingKey)) {
			return (T) rabbitTemplate.convertSendAndReceive(exchange, routingKey, message);
		} else if (StringUtils.isNotEmpty(routingKey)) {
			return (T) rabbitTemplate.convertSendAndReceive(routingKey, message);
		} else {
			return (T) rabbitTemplate.convertSendAndReceive(message);
		}
	}

	public static class PendingConfirm {

		private volatile String savedCorrelation;
		private final LinkedBlockingQueue<Boolean> queue;

		public PendingConfirm() {
			this.queue = new LinkedBlockingQueue<Boolean>();
		}

		public PendingConfirm(String savedCorrelation) {
			this();
			this.setSavedCorrelation(savedCorrelation);
		}

		public String getSavedCorrelation() {
			return savedCorrelation;
		}

		public void setSavedCorrelation(String savedCorrelation) {
			this.savedCorrelation = savedCorrelation;
		}

		public LinkedBlockingQueue<Boolean> getQueue() {
			return queue;
		}
	}

}
